package com.aotain.jupiter.ipstat;

import java.util.ArrayList;
import java.util.Collection;
import java.util.HashMap;

import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.java.tuple.Tuple2;

import com.aotain.jupiter.util.CommonUtil;

public class IPFlowStatMap implements MapFunction<Tuple2<String,String>,Tuple2<String,IPStatTuple>>{

	/**
	 * serialVersionUID
	 */
	private static final long serialVersionUID = 2194827863697299819L;

	@Override
	public Tuple2<String,IPStatTuple> map(Tuple2<String, String> arg0)
			throws Exception {
		// TODO Auto-generated method stub
//		//dip+ "|" +dport+ "|" +sip+ "|" +sport+ "|" +gisstrBase64+ "|" 
//		//+ domainname+ "|" +url+ "|" +type+ "|" +delaytime+ "|" +idcno+ "|" 
//		// +upStreamOctets+ "|" +upStreamPacket+ "|" +status+ "|" +userAgent+ "|" +cookie+ "|" 
//		// +dnStreamOctets+ "|" +dnStreamPacket+ "|" +refer
		
		Tuple2<String,IPStatTuple> ret = new Tuple2<String,IPStatTuple>();
		String logString = arg0.f1;
		String[] items = logString.split("\\|",-1);
		if(items.length <= 10)
			return ret;
		String sip = arg0.f0;
		
		IPStatTuple tuple = new IPStatTuple();
		tuple.setSourceIP(sip);
		
		
		long upStreamOctets = 0; //UPSTREAMOCTETS
		long upStreamPacket = 0; //UPSTREAMPACKET
		long dnStreamOctets = 0; //DNSTREAMOCTETS
		long dnStreamPacket = 0; //DNSTREAMPACKET
		
		upStreamOctets = Long.parseLong(items[10]);
		upStreamPacket = Long.parseLong(items[11]);
		dnStreamOctets = Long.parseLong(items[15]);
		dnStreamPacket = Long.parseLong(items[16]);
		
		HashMap<String,Long> valueMap =new HashMap<String,Long>();
		valueMap.put(CommonUtil.UPSTREAMOCTETS, upStreamOctets);
		valueMap.put(CommonUtil.UPSTREAMPACKET, upStreamPacket);
		valueMap.put(CommonUtil.DNSTREAMOCTETS, dnStreamOctets);
		valueMap.put(CommonUtil.DNSTREAMPACKET, dnStreamPacket);
		
		tuple.setValueMap(valueMap);
	
		ret.setFields(sip, tuple);
		return ret;
	}

	

}
